package com.chukun.flink.stream.state.key;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;

/**
 * @author chukun
 * @version 1.0.0
 * @description 带状态的flatmap操作符 listState的基本使用
 * @createTime 2022年05月15日 22:59:00
 */
public class ListStateFlatMap extends RichFlatMapFunction<Tuple2<Integer, Integer>, String> {

    private static final Logger logger = LoggerFactory.getLogger(ListStateFlatMap.class);

    // 创建ListState类型的状态结构
    private transient ListState<Tuple2<Integer,Integer>> listState;

    @Override
    public void flatMap(Tuple2<Integer, Integer> input, Collector<String> collector) throws Exception {
        // 将当前元素加入到listState中
        listState.add(input);

        Iterator<Tuple2<Integer, Integer>> iterator = listState.get().iterator();
        int number = 0;
        StringBuffer buffer = new StringBuffer();
        while (iterator.hasNext()) {
            // 遍历状态中的值，将值拼接到变量buffer的字符串中
            buffer.append("-->" + iterator.next());
            number ++;
            //当前的状态中存储了3个元素，便向下游操作符输出该状态的所有元素
            if (number == 3) {
                // 清空状态集合
                listState.clear();
                // 向下游操作符输出该状态的所有元素
                collector.collect(buffer.toString());

            }
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        logger.info("{},{}", Thread.currentThread().getName(), "恢复或初始化状态");

        // 创建一个ListStateDescriptor状态描述符，指定名称为ListStateFlatMap，类型为 Tuple2<Integer, Integer>
        ListStateDescriptor<Tuple2<Integer,Integer>> descriptor = new ListStateDescriptor<>(
                "ListStateFlatMap",
                TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {}));
        // 初始化valueState
        listState = getRuntimeContext().getListState(descriptor);
        logger.info("初始化 listState ： {}", listState);
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 封装了数据源函数，以及keyBy操作的数据流
        KeyedStream<Tuple2<Integer, Integer>, Integer> keyedStream = KeyStateBase.before(env);

        // 在keyedStream中使用有状态的FlatMap操作符
        DataStream<String> resultStream = keyedStream.flatMap(new ListStateFlatMap());

        resultStream.print("输出结果 : ");

        env.execute("ListStateFlatMap");
    }
}
